package com.zhkc.wisdom.common.iot.service;

import cn.hutool.core.collection.CollectionUtil;
import com.zhkc.wisdom.common.iot.model.TimeSeriesModel;
import com.zhkc.wisdom.common.iot.properties.IotProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class IotDBService {

    private SessionPool sessionPool;
    private IotProperties iotProperties;

    public IotDBService(SessionPool sessionPool, IotProperties iotProperties) {
        this.sessionPool = sessionPool;
        this.iotProperties = iotProperties;
    }

    /**
     * 创建数据库
     *
     * @param database 数据库名称 例如 root.ln
     */
    public void createDatabase(String database) throws IoTDBConnectionException, StatementExecutionException {
        sessionPool.createDatabase(database);
    }

    /**
     * 创建对齐序列
     *
     * @param deviceId   设备（传入具体的设备路径，例如：root.ln.deviceId001）
     * @param timeSeries 序列list对象
     */
    public void createAlignedTimeSeries(String deviceId, List<TimeSeriesModel> timeSeries) throws IoTDBConnectionException, StatementExecutionException {
        if (CollectionUtil.isEmpty(timeSeries)) {
            return;
        }
        //测量项列表，指定了要创建的时间序列的具体测量项
        List<String> measurements = new ArrayList<>();
        //数据类型列表，每个测量项对应的数据类型（如整型、浮点型等)
        List<TSDataType> dataTypes = new ArrayList<>();
        //编码类型列表，指定如何编码每个测量项的数据
        List<TSEncoding> encodings = new ArrayList<>();
        //压缩类型列表，指定如何压缩每个测量项的数据
        List<CompressionType> compressors = new ArrayList<>();
        //测量项别名列表，为每个测量项提供一个可选的别名
        List<String> measurementAliasList = new ArrayList<>();

        timeSeries.forEach(time -> {
            measurements.add(time.getSeries());
            dataTypes.add(time.getDataType());
            encodings.add(time.getEncoding());
            compressors.add(time.getCompressor());
            measurementAliasList.add(time.getAlias());
        });
        sessionPool.createAlignedTimeseries(deviceId, measurements, dataTypes, encodings, compressors, measurementAliasList);
    }

    /**
     * 删除时间序列(单路径)
     */
    public void deleteTimeSeries(String path) throws IoTDBConnectionException, StatementExecutionException {
        sessionPool.deleteTimeseries(path);
    }

    /**
     * 删除时间序列(多路径)
     */
    public void deleteTimeSeries(List<String> paths) throws IoTDBConnectionException, StatementExecutionException {
        sessionPool.deleteTimeseries(paths);
    }


    /**
     * 检测时间序列是否存在
     */
    public boolean checkTimeSeriesExists(String path) throws IoTDBConnectionException, StatementExecutionException {
        return sessionPool.checkTimeseriesExists(path);
    }


    /**
     * Tablet 数据写入(单个)
     */
    public void insertAlignedTablet(Tablet tablet) throws IoTDBConnectionException, StatementExecutionException {
        sessionPool.insertAlignedTablet(tablet);
    }

    /**
     * Tablet 数据写入(多个)
     */
    public void insertAlignedTablets(Map<String, Tablet> tablets) throws IoTDBConnectionException, StatementExecutionException {
        sessionPool.insertAlignedTablets(tablets);
    }

    /**
     * @param deviceId 设备id（对应路径上的设备标识）
     * @param time     数据采集时间（时间戳）
     * @param factors  设备的因子（系统配置的该设备的监测因子（统一格式 c_xxxxxx_avg、c_xxxxxx_rtd））
     * @param map      数采仪上传的数据（212协议解析之后的只包含数据段的map）
     * @throws IoTDBConnectionException
     * @throws StatementExecutionException
     */
    public void insertAlignedTabletBy212(String deviceId, String time, List<String> factors, Map<String, Map<String, String>> map) throws IoTDBConnectionException, StatementExecutionException {
        //先把map处理成"因子_avg"这种形式
        Map<String, Object> resultMap = new HashMap<>();
        map.forEach((key, value) -> {
            if (value != null) {
                //这里的map的value格式是一个json字符串，所以再次转换
                value.forEach((vk, va) -> {
                    resultMap.put("c_" + key + "_" + vk.toLowerCase(), va);
                });
            }
        });

        List<MeasurementSchema> schemas = new ArrayList<>();
        //组装tablet,因为单条数据直接插入，maxRowNumber = 1
        factors.forEach(factor -> {
            schemas.add(new MeasurementSchema(factor, TSDataType.TEXT, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED));

        });
        Tablet tablet = new Tablet(iotProperties.getDatabase() + "." + deviceId, schemas, 1);
        tablet.addTimestamp(0, Long.parseLong(time));
        int i = tablet.rowSize++;
        factors.forEach(factor -> {
            tablet.addValue(factor, i, resultMap.get(factor));
        });

        sessionPool.insertAlignedTablet(tablet);
    }


    /**
     * 时间序列原始数据范围查询，指定的查询时间范围为左闭右开区间，包含开始时间但不包含结束时间；
     * timeout超时（毫秒）
     */
    public List<Map<String, String>> executeRawDataQuery(List<String> paths, long startTime, long endTime, long timeout) throws IoTDBConnectionException, StatementExecutionException {
        SessionDataSetWrapper wrapper = sessionPool.executeRawDataQuery(paths, startTime, endTime, timeout);
        SessionDataSet dataSet = wrapper.getSessionDataSet();
        return mappingResultSet(dataSet);
    }

    /**
     * 直接执行查询语句
     */
    public List<Map<String, String>> executeQueryStatement(String sql) throws IoTDBConnectionException, StatementExecutionException {
        SessionDataSetWrapper wrapper = sessionPool.executeQueryStatement(sql);
        SessionDataSet dataSet = wrapper.getSessionDataSet();
        return mappingResultSet(dataSet);
    }


    private List<Map<String, String>> mappingResultSet(SessionDataSet dataSet) throws StatementExecutionException, IoTDBConnectionException {
        List<String> columnNames = dataSet.getColumnNames();
        List<Map<String, String>> resultList = new ArrayList<>();
        while (dataSet.hasNext()) {
            RowRecord rowRecord = dataSet.next();
            List<Field> fields = rowRecord.getFields();
            Map<String, String> map = new HashMap<>();
            //time表示数据采集时间
            map.put("time", String.valueOf(rowRecord.getTimestamp()));
            for (int i = 0; i < fields.size(); i++) {
                map.put(columnNames.get(i + 1), fields.get(i).getStringValue());
            }
            resultList.add(map);
        }
        return resultList;
    }
}




